package com.bigdata.flink.datastream;

import com.bigdata.flink.util.StreamEnvUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class OperatorFunctionTest {
  public static void main(String[] args) throws Exception {
    reduce();
  }

  // 通过key分组，对每组的输入、来一个计算一个得到一个结果打印、并作为value1循环输入
  public static void reduce() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    KeyedStream<String, String> keyedStream = socketSource.keyBy(new KeySelector<String, String>() {
      @Override
      public String getKey(String value) throws Exception {
        return value.split("\\s")[0];
      }
    });

    SingleOutputStreamOperator<String> reduce = keyedStream.reduce(new ReduceFunction<String>() {
      @Override
      public String reduce(String value1, String value2) throws Exception {
        String[] split1 = value1.split("\\s");
        String[] split2 = value2.split("\\s");
        int sum = Integer.parseInt(split1[1]) + Integer.parseInt(split2[1]);
        return split1[0] + " " + sum;
      }
    });

    reduce.print();
    env.execute();
  }

  public static void flatMap() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = socketSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
          // 把输入的值进行处理，输出到固定格式
          @Override
          public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] words = value.split("\\s");
            for (String word : words) {
              out.collect(Tuple2.of(word, 1));
            }
          }
        }).returns(Types.TUPLE(Types.STRING, Types.INT))
        .keyBy(f -> f.f0)
        .sum(1);


    flatMap.print();
    env.execute();
  }


  public static void map() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    SingleOutputStreamOperator<Integer> map = socketSource.map(new MapFunction<String, Integer>() {
      @Override
      public Integer map(String value) throws Exception {
        return Integer.parseInt(value) * 2;
      }
    });

    map.print();
    env.execute();
  }

}
